package com.to8to.kitt.esb;

import java.io.File;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import sun.misc.Signal;
import sun.misc.SignalHandler;

import com.to8to.kitt.Constants;
import com.to8to.kitt.utils.LogUtils;
import com.to8to.kitt.utils.NettyUtils;



public abstract class AbstractServerDeamon extends AbstractDeamon{

	public static Logger logger = LoggerFactory.getLogger(AbstractServerDeamon.class);  

	protected String bindHost;
	protected int bindPort;
	protected boolean isBind = false;
	protected boolean randomPort = false;
//	protected int maxIOThreads = Constants.MAX_IO_THREADS; ////<<<<<<<
	
	protected ServerBootstrap bootstrap;
	protected ChannelFuture channelFuture;
	protected ChannelInitializer<Channel> channelInitilizer;
	
	protected EventLoopGroup bossGroup = new NioEventLoopGroup(1);
	
	protected EventLoopGroup workerGroup = new NioEventLoopGroup(Constants.MAX_IO_THREADS);
	
	protected Thread reportScanThread = new Thread(new Runnable(){
		@Override
		public void run() {
			
			logger.debug("启动监听上报线程 ...");
			
			int reportNum = 0;
			
			long lastReportTime=System.currentTimeMillis();
			
			
			while(enableReport)
			{
				try {
					boolean isok = ensureReport();
					
					if(!isok)
					{
						long now = System.currentTimeMillis();
						
						if(now - lastReportTime < 30000)
						{
							reportNum ++;
							
							if(reportNum >= 3)
							{
								System.out.println("频繁失联上报，超过3次，自杀!!!!");
								AbstractServerDeamon.this.exitForSignle15();
							}
						}
						else 
						{
							reportNum = 0;
						}
						
						lastReportTime = now;
					}
					
					Thread.sleep(reportScanIntervalMS);
				} catch (Throwable e) {
					
					NettyUtils.exitIfOutOfMemory(e);
					
					e.printStackTrace();
				}
			}
			
			logger.warn("终止zk监听!");
			
		}});
	
	protected void report(boolean unreg)
	{
		logger.info("<<<< report my address to ZK .... start! >>>>");
		try {
			
			if(serviceInstance != null && unreg)
				serviceDiscovery.unregisterService(serviceInstance);
			
//			if(serviceInstance == null || newinstance)
				serviceInstance = ServiceInstance.<String>builder().id(instance).name(service).address(reportHost).payload(payload).port(reportPort).build();
			
			String dir = System.getProperty("java.run.dir");
			
			String prog = System.getProperty("java.apps.prog");
			
			if(!StringUtils.isEmpty(dir) && !StringUtils.isEmpty(prog))
			{
				File file = new File(dir + "/" + prog + ".zk");
				String pid = NettyUtils.getPID();
				FileUtils.writeStringToFile(file, pid + "\n" +  ZK.ZK_CONFIG.servicePath + "/" + service + "/" + instance);
			}
			
			serviceDiscovery.registerService(serviceInstance);
			
			if(!reportScanThread.isAlive())
				reportScanThread.start();
			
			logger.info("<<<< report my address to ZK .... success! >>>>");
			
		} catch (Exception e) {
			e.printStackTrace();
			logger.error(LogUtils.logStackTrace(e));
			logger.info("############# report my address to ZK .... failed! ##############");
		}
	}
	protected boolean ensureReport()
	{
		String path = ZK.ZK_CONFIG.servicePath + "/" + service + "/" + instance;
//		logger.trace("ensureReport: {}", path);
		try {
			Stat stat = ZK.ZK_CLIENT.checkExists().forPath(path);
			
//			logger.trace("stat: {}", stat);
			
			if(stat == null)
			{
				logger.warn("与zk汇报失联,重新上报!!!!!!");
				report(true);
				return false;
			}
			else 
			{
				return true;
			}
		} catch (Throwable e) {
			
			NettyUtils.exitIfOutOfMemory(e);
			
			logger.error(LogUtils.logStackTrace(e));
			e.printStackTrace();
		}
		
		return false;
	}
	
	public AbstractServerDeamon()
	{
		
	}
	
	public AbstractServerDeamon(ChannelInitializer<Channel> channelInitilizer)
	{
		this.channelInitilizer = channelInitilizer;
	}

//	public int getMaxIOThreads() {
//		return maxIOThreads;
//	}
//
//	public void setMaxIOThreads(int maxIOThreads) {
//		this.maxIOThreads = maxIOThreads;
//	}

	public ChannelInitializer<Channel> getChannelInitilizer() {
		return channelInitilizer;
	}

	public void setChannelInitilizer(ChannelInitializer<Channel> channelInitilizer) {
		this.channelInitilizer = channelInitilizer;
	}
	
	
	public String getBindHost() {
		return bindHost;
	}

	public void setBindHost(String bindHost) {
		this.bindHost = bindHost;
	}


	
	public int getBindPort() {
		return bindPort;
	}

	public void setBindPort(int bindPort) {
		this.bindPort = bindPort;
	}

	protected boolean isAvailable() 
	{
		return true;
	}
	
	
	protected void bindAndReport(String bindHost, int bindPort, String reportHost, int reportPort)
	{
		boolean rebind = false;
		if(!bindHost.equals(this.bindHost) || this.bindPort != bindPort)
			rebind = true;
		
		boolean rereport = false;
		if(!reportHost.equals(this.reportHost) || this.reportPort != reportPort)
			rereport = true;
		
		this.bindHost = bindHost;
		this.bindPort = bindPort;
		this.reportHost = reportHost;
		this.reportPort = reportPort;
		
		if(rebind)
			bind();
		
		if(rereport)
			report(true);
	}
	
	protected boolean bind()
	{
		logger.info("try to bind address - " + bindHost + ":" + bindPort);
		
		if(bindHost == null || bindPort == 0)
		{
			logger.error("please set bindHost, bindPort first!");
			return false;
		}
		
		
		try {
			
			if(channelFuture != null && channelFuture.channel() != null)
			{
				logger.info("close existing channel: " + channelFuture.channel());
				channelFuture.channel().close();
			}
			
			
			for(int i = 0; i < 10; ++ i)
			{
				try
				{
					if(bindHost != null)
						channelFuture = bootstrap.bind(bindHost, bindPort).sync();
					else 
						channelFuture = bootstrap.bind(bindPort).sync();
					
					break;
				}
				catch(Exception e)
				{
					if(randomPort)
						bindPort = (int)System.currentTimeMillis() % 5000 + 35000;
				}
			}

//			logger.info("bind channel: {}" + channelFuture.channel());
			
			logger.info("bind success: {}:{}" , bindHost, bindPort);
			
			isBind = true;
			
			return true;
		
		} catch (Exception e) {
			
			e.printStackTrace();
			
			logger.error(LogUtils.logStackTrace(e));
		}
		
		System.out.println("bind failed!!!!!");
		
		logger.error("bind failed!");
		
		return false;
	}
	
	
	private boolean initServerBootstrap()
	{
		logger.info("<<<<<<< initServerBootstrap() ... start! >>>>>>");
		if(channelInitilizer == null)
		{
			logger.error("channalInitializer is null, please set first!");
			return false;
		}
		
//		EventLoopGroup bossGroup = new NioEventLoopGroup(1);
//		EventLoopGroup workerGroup = new NioEventLoopGroup(nthreads);
        
        try {
            bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, Constants.SO_BACKLOG)  //requested maximum length of the queue of incoming connections.
             .childOption(ChannelOption.SO_SNDBUF, Constants.SO_SNDBUF)
             .childOption(ChannelOption.SO_RCVBUF, Constants.SO_RCVBUF)
             .childHandler(channelInitilizer);
            
            logger.info("<<<<<<< initServerBootstrap() ... success! >>>>>>>");
            
            return true;
        } catch (Exception e) {
        	logger.error(LogUtils.logStackTrace(e));
		} 
        
        logger.info("#################  initServerBootstrap() ... failed! ##################");
        return false;
	}
	
	
//	protected void report()
//	{
//		if(!isBind)
//		{
//			logger.error("server address is not bind! please bind first!");
//			return ;
//		}
//		
//		try {
//			
//			if(serviceInstance != null)
//				serviceDiscovery.unregisterService(serviceInstance);
//			
//			initReportInstance();
//			
//			serviceDiscovery.registerService(serviceInstance);
//			
//		} catch (Exception e) {
//			logger.error(LogUtils.logStackTrace(e));
//		}
//	}
	
	protected void exitForSignle15(){
		System.out.println("进程被强迫终止 ...");
		System.out.println("关闭监听端口...");
		enableReport = false;
		 
		if(channelFuture != null && channelFuture.channel() != null)
		{
			Channel channel = channelFuture.channel();
			logger.info("close existing channel: " + channel);
			ChannelFuture cf = channel.close();
			cf.addListener(new ChannelFutureListener(){
				@Override
				public void operationComplete(ChannelFuture future)
						throws Exception {
					if(future.isSuccess())
					{
						System.out.println("端口关闭成功!");
					}
					else 
						System.out.println("端口关闭失败!");
				}});
		}
		 
		 if(serviceInstance != null)
		 {
			System.out.println("注销zk实例!");
			 
			try {
				serviceDiscovery.unregisterService(serviceInstance);
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		 }
		 
		try {
			Thread.sleep(Constants.KILL_TIMEWAIT);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		System.out.println("进程退出!!!!!!!"); 
		System.exit(-1);
	}
	
	@SuppressWarnings("restriction")
	public boolean init()
	{
		if(!initServerBootstrap())
			return false;
		
		if(!super.init())
			return false;

		if(!syncConfigFromZK) //不从 zk读配置，则程序手动配置绑定端口和ip
		{
			if(!bind())
				return false;
		}

		while(!isBind)
		{
			try {
				Thread.sleep(3000);
			} catch (InterruptedException e) {
				logger.error(LogUtils.logStackTrace(e));
			}
		}
		
		System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
		System.out.println("@@@@@@@@@@@@@@@@@@@@@              updated on:" + Constants.ESB_VERSION+"          @@@@@@@@@@@@@@@@@@@");
		System.out.println("@@@@@@@@@@@@@@@@@@@@@  server bind success: " + bindHost + ":" + bindPort + "  @@@@@@@@@@@@@@@@@@@@");
		System.out.println("@@@@@@@@@@@@@@@@@@@@@                                            @@@@@@@@@@@@@@@@@@@");
		System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
		
		Signal.handle(new Signal("TERM"), new SignalHandler(){
			 @Override  
			 public void handle(Signal signal) {  
				 AbstractServerDeamon.this.exitForSignle15();
			 }
		});  
		
		return true;
	}
}
